package com.example.dayu521.sg_oss.service.impl;

import com.example.dayu521.sg_oss.domain.NumSplit;
import com.example.dayu521.sg_oss.service.OrderNumSource;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Component
@ConditionalOnProperty(prefix = "sg.num",name = "backend",havingValue = "redis-cluster")
public class RedisClusterSourceImpl implements OrderNumSource {

    @Resource(name = "redisTemplateCluster")
    private RedisTemplate<String, Object> redisTemplateCluster;

    @Resource(name = "redisTemplateCluster")
    private RedisTemplate<String, NumSplit> rrr;


    private final AtomicLong atomicLong = new AtomicLong(-1);
    private Long node_size;

    @Resource(name = "init_node_map")
    private Map<Integer, Integer> solt_map;

    private static final String str_format = "{%d}.%s";

    @PostConstruct
    public void init_node_size() {
        node_size = (long) solt_map.size();
    }

    @Override
    public Long get_random_node() {

        return (long) solt_map.get((int) (atomicLong.incrementAndGet() % node_size));
    }

    @Override
    public void save_to(Long node_index, Map<String, ? extends Object> multi_keys) {
        var s = new HashMap<String, Object>();
        multi_keys.entrySet().forEach(i -> {
            s.put(String.format(str_format, node_index, i.getKey()), i.getValue());
        });
        redisTemplateCluster.opsForValue().multiSet(s);
    }

    @Override
    public <T> void save_to(Long node_index, String key, T obj,long version) {
        redisTemplateCluster.execute(RedisScript.of("""
                local current_version = redis.call('GET', KEYS[2])
                if current_version > ARGV[2]
                  then return false
                end
                redis.call('SET', KEYS[1], ARGV[1])
                redis.call('SET', KEYS[2], ARGV[2])
                return true
                """,Boolean.class), List.of(String.format(str_format, node_index, key),String.format(str_format, node_index, key+":version")),obj,version);
        redisTemplateCluster.opsForValue().set(String.format(str_format, node_index, key), obj);
    }

    @Override
    public <T> boolean save_not_exist(Long node_index, String key, T obj) {
        return redisTemplateCluster.opsForValue().setIfAbsent(String.format(str_format, node_index, key), obj);
    }

    @Override
    public boolean save_not_exist(Long node_index, Map<String, ?> multi_keys) {
        var s = new HashMap<String, Object>();
        multi_keys.entrySet().forEach(i -> {
            s.put(String.format(str_format, node_index, i.getKey()), i.getValue());
        });
        return redisTemplateCluster.opsForValue().multiSetIfAbsent(s);
    }

    @Override
    public <T> T pull_from(Long node_index, String key) {
//        Map<String, Object> entries = rtc_hash.entries(String.format(str_format, node_index, key));
//        redisTemplateCluster.boundValueOps()
        var xx = rrr.opsForValue().get(String.format(str_format, node_index, key));
        return (T) redisTemplateCluster.opsForValue().get(String.format(str_format, node_index, key));
//        if(entries.isEmpty()){
//        }
//        return (T) RedisHashBe.map2Obj(entries);
    }

    @Override
    public Long get_next_long(Long node_index, String key) {
        return redisTemplateCluster.opsForValue().decrement(String.format(str_format, node_index, key));
    }
}
